package com.jvup.common.concurrent;

import android.os.Handler;
import android.os.Looper;

import com.jvup.common.function.Function;

/**
 * 被观察者
 */
public abstract class Observed<T> {

    private static Handler handler = new Handler(Looper.myLooper());

    public abstract void subscribe(Observer<T> observer);

    public static <T> Observed<T> create(Observed<T> observed) {
        return observed;
    }

    public static <T> Observed<T> just(T... data) {
        return new Observed<T>() {
            @Override
            public void subscribe(Observer<T> observer) {
                if(data != null) {
                    for(T value : data) {
                        observer.onNext(value);
                    }
                }
            }
        };
    }

    public Observed<T> flatMap() {
        return new Observed<T>() {
            @Override
            public void subscribe(final Observer<T> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        observer.onNext(value);
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    public <R> Observed<R> map(final Function<T, R> function) {
        return new Observed<R>() {
            @Override
            public void subscribe(Observer<R> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        observer.onNext(function.apply(value));
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    public Observed<T> subscribeOn() {
        return new Observed<T>() {
            @Override
            public void subscribe(final Observer<T> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        new Thread(() -> observer.onNext(value)).start();
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    public Observed<T> subscribeBy() {
        return new Observed<T>() {
            @Override
            public void subscribe(final Observer<T> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        handler.post(() -> observer.onNext(value));
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }
}
